package com.shujia.flink.core;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo03Parallelism {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
         * Flink指定并行度的方式：
         * 1、在代码中指定，通过env设置并行度即可
         * 2、在提交任务时指定：（推荐）
         *        2.1 在Yarn-Session模式的WEB界面提交时可以通过输入框指定
         *        2.2 在命令行通过命令提交，通过-p/--parallelism参数设置并行度
         * 3、每个DS进行操作时可以单独设置并行度
         */

        /*
         * 如何确定要几个并行度？
         * 通过数据的处理速度来确定：
         *      实时任务没有聚合操作：1w~10w/s/并行度
         *      实时任务有聚合操作：1000~1w/s/并行度
         * 最终并行度会影响任务占用的Slot槽位数量：一个并行度就需要一个TaskSlot
         * Task任务的数量会受到并行度的影响：并行度改变或者是有KeyBy类操作的时候会出现多个Task

         */
//        env.setParallelism(2);

        DataStream<String> lineDS = env.socketTextStream("master", 8888)
                .setParallelism(1); // SocketSource默认且只能是1个并行度，不能改变

        DataStream<String> wordsDS = lineDS
                .flatMap((line, out) -> {
                    for (String word : line.split(",")) {
                        out.collect(word);
                    }
                }, Types.STRING)
                .setParallelism(4);

        DataStream<Tuple2<String, Integer>> wordsKVDS = wordsDS
                .map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT))
                .setParallelism(4);

        DataStream<Tuple2<String, Integer>> wordCntDS = wordsKVDS
                .keyBy(kv -> kv.f0, Types.STRING)
                .sum(1)
                .setParallelism(4);

        wordCntDS
                .print(); // 默认只有一个并行度

        // 启动流处理任务
        env.execute();


    }


}
